package com.flink.hbase.statemanagement.managed;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * ListState托管状态示例
 * 
 * 展示如何使用ListState进行状态管理：
 * 1. 用户行为轨迹追踪
 * 2. 滑动窗口数据缓存
 * 3. 购物车商品管理
 * 4. 消息队列状态管理
 * 
 * @author Flink Team
 */
public class ListStateExample {
    
    /**
     * 用户行为事件
     */
    public static class UserBehaviorEvent {
        public String userId;
        public String action;
        public String page;
        public String product;
        public long timestamp;
        
        public UserBehaviorEvent() {}
        
        public UserBehaviorEvent(String userId, String action, String page, String product, long timestamp) {
            this.userId = userId;
            this.action = action;
            this.page = page;
            this.product = product;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("UserBehaviorEvent{userId='%s', action='%s', page='%s', product='%s', timestamp=%d}", 
                userId, action, page, product, timestamp);
        }
    }
    
    /**
     * 购物车商品
     */
    public static class CartItem {
        public String productId;
        public String productName;
        public int quantity;
        public double price;
        public long addTime;
        
        public CartItem() {}
        
        public CartItem(String productId, String productName, int quantity, double price, long addTime) {
            this.productId = productId;
            this.productName = productName;
            this.quantity = quantity;
            this.price = price;
            this.addTime = addTime;
        }
        
        @Override
        public String toString() {
            return String.format("CartItem{productId='%s', productName='%s', quantity=%d, price=%.2f, addTime=%d}", 
                productId, productName, quantity, price, addTime);
        }
    }
    
    /**
     * 消息事件
     */
    public static class MessageEvent {
        public String userId;
        public String messageId;
        public String content;
        public String priority;
        public long timestamp;
        
        public MessageEvent() {}
        
        public MessageEvent(String userId, String messageId, String content, String priority, long timestamp) {
            this.userId = userId;
            this.messageId = messageId;
            this.content = content;
            this.priority = priority;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return String.format("MessageEvent{userId='%s', messageId='%s', content='%s', priority='%s', timestamp=%d}", 
                userId, messageId, content, priority, timestamp);
        }
    }
    
    /**
     * 用户行为轨迹追踪函数
     * 使用ListState存储用户最近的行为记录
     */
    public static class UserBehaviorTracker extends RichFlatMapFunction<UserBehaviorEvent, Tuple2<String, List<UserBehaviorEvent>>> {
        
        // 用户行为历史状态
        private ListState<UserBehaviorEvent> behaviorHistoryState;
        
        private static final int MAX_BEHAVIOR_HISTORY = 10; // 最多保存10条行为记录
        
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建ListState描述符
            ListStateDescriptor<UserBehaviorEvent> behaviorHistoryDescriptor = new ListStateDescriptor<>(
                "behavior-history", // 状态名称
                TypeInformation.of(UserBehaviorEvent.class) // 状态类型
            );
            
            // 获取状态
            behaviorHistoryState = getRuntimeContext().getListState(behaviorHistoryDescriptor);
        }
        
        @Override
        public void flatMap(UserBehaviorEvent event, Collector<Tuple2<String, List<UserBehaviorEvent>>> collector) throws Exception {
            // 获取当前行为历史
            List<UserBehaviorEvent> behaviorHistory = new ArrayList<>();
            for (UserBehaviorEvent behavior : behaviorHistoryState.get()) {
                behaviorHistory.add(behavior);
            }
            
            // 添加新行为
            behaviorHistory.add(event);
            
            // 保持最新的N条记录
            if (behaviorHistory.size() > MAX_BEHAVIOR_HISTORY) {
                behaviorHistory.remove(0);
            }
            
            // 清空状态并重新添加
            behaviorHistoryState.clear();
            behaviorHistoryState.addAll(behaviorHistory);
            
            // 输出结果
            collector.collect(new Tuple2<>(event.userId, new ArrayList<>(behaviorHistory)));
            
            System.out.println(String.format("📊 用户 %s 行为轨迹更新: %s，历史记录数: %d", 
                event.userId, event.action, behaviorHistory.size()));
        }
    }
    
    /**
     * 滑动窗口数据缓存函数
     * 使用ListState实现自定义滑动窗口
     */
    public static class SlidingWindowCache extends KeyedProcessFunction<String, UserBehaviorEvent, Tuple3<String, String, Integer>> {
        
        // 窗口数据缓存状态
        private ListState<UserBehaviorEvent> windowDataState;
        
        private static final long WINDOW_SIZE = 30 * 1000L; // 30秒窗口
        private static final long SLIDE_SIZE = 10 * 1000L; // 10秒滑动
        
        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<UserBehaviorEvent> windowDataDescriptor = new ListStateDescriptor<>(
                "window-data", TypeInformation.of(UserBehaviorEvent.class));
            
            windowDataState = getRuntimeContext().getListState(windowDataDescriptor);
        }
        
        @Override
        public void processElement(UserBehaviorEvent event, Context context, 
                                 Collector<Tuple3<String, String, Integer>> collector) throws Exception {
            
            // 获取当前窗口数据
            List<UserBehaviorEvent> windowData = new ArrayList<>();
            for (UserBehaviorEvent data : windowDataState.get()) {
                windowData.add(data);
            }
            
            // 添加新数据
            windowData.add(event);
            
            // 清理过期数据
            long currentTime = context.timerService().currentProcessingTime();
            windowData.removeIf(data -> currentTime - data.timestamp > WINDOW_SIZE);
            
            // 更新状态
            windowDataState.clear();
            windowDataState.addAll(windowData);
            
            // 计算各种行为的统计
            int viewCount = 0, clickCount = 0, purchaseCount = 0;
            for (UserBehaviorEvent data : windowData) {
                switch (data.action) {
                    case "view":
                        viewCount++;
                        break;
                    case "click":
                        clickCount++;
                        break;
                    case "purchase":
                        purchaseCount++;
                        break;
                }
            }
            
            // 输出统计结果
            collector.collect(new Tuple3<>(event.userId, "view", viewCount));
            collector.collect(new Tuple3<>(event.userId, "click", clickCount));
            collector.collect(new Tuple3<>(event.userId, "purchase", purchaseCount));
            
            System.out.println(String.format("📈 用户 %s 滑动窗口统计: 浏览=%d, 点击=%d, 购买=%d", 
                event.userId, viewCount, clickCount, purchaseCount));
            
            // 注册定时器进行定期清理
            context.timerService().registerProcessingTimeTimer(currentTime + SLIDE_SIZE);
        }
        
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, 
                           Collector<Tuple3<String, String, Integer>> out) throws Exception {
            
            // 定期清理过期数据
            List<UserBehaviorEvent> windowData = new ArrayList<>();
            for (UserBehaviorEvent data : windowDataState.get()) {
                if (timestamp - data.timestamp <= WINDOW_SIZE) {
                    windowData.add(data);
                }
            }
            
            // 更新状态
            windowDataState.clear();
            windowDataState.addAll(windowData);
            
            System.out.println("🧹 定时清理过期数据: " + ctx.getCurrentKey());
        }
    }
    
    /**
     * 购物车商品管理函数
     * 使用ListState管理购物车商品
     */
    public static class ShoppingCartManager extends KeyedProcessFunction<String, CartItem, Tuple2<String, List<CartItem>>> {
        
        // 购物车商品状态
        private ListState<CartItem> cartItemsState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<CartItem> cartItemsDescriptor = new ListStateDescriptor<>(
                "cart-items", TypeInformation.of(CartItem.class));
            
            cartItemsState = getRuntimeContext().getListState(cartItemsDescriptor);
        }
        
        @Override
        public void processElement(CartItem newItem, Context context, 
                                 Collector<Tuple2<String, List<CartItem>>> collector) throws Exception {
            
            // 获取当前购物车商品
            List<CartItem> cartItems = new ArrayList<>();
            for (CartItem item : cartItemsState.get()) {
                cartItems.add(item);
            }
            
            // 检查商品是否已存在
            boolean itemExists = false;
            for (int i = 0; i < cartItems.size(); i++) {
                CartItem existingItem = cartItems.get(i);
                if (existingItem.productId.equals(newItem.productId)) {
                    // 更新数量
                    existingItem.quantity += newItem.quantity;
                    itemExists = true;
                    break;
                }
            }
            
            // 如果商品不存在，添加新商品
            if (!itemExists) {
                cartItems.add(newItem);
            }
            
            // 移除数量为0的商品
            cartItems.removeIf(item -> item.quantity <= 0);
            
            // 更新状态
            cartItemsState.clear();
            cartItemsState.addAll(cartItems);
            
            // 输出结果
            collector.collect(new Tuple2<>(newItem.productId, new ArrayList<>(cartItems)));
            
            // 计算购物车总价
            double totalPrice = cartItems.stream()
                .mapToDouble(item -> item.price * item.quantity)
                .sum();
            
            System.out.println(String.format("🛒 购物车更新: 商品数=%d, 总价=%.2f", 
                cartItems.size(), totalPrice));
        }
    }
    
    /**
     * 消息队列状态管理函数
     * 使用ListState实现消息队列功能
     */
    public static class MessageQueueManager extends KeyedProcessFunction<String, MessageEvent, Tuple2<String, String>> {
        
        // 消息队列状态
        private ListState<MessageEvent> messageQueueState;
        
        private static final int MAX_QUEUE_SIZE = 100; // 最大队列大小
        
        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<MessageEvent> messageQueueDescriptor = new ListStateDescriptor<>(
                "message-queue", TypeInformation.of(MessageEvent.class));
            
            messageQueueState = getRuntimeContext().getListState(messageQueueDescriptor);
        }
        
        @Override
        public void processElement(MessageEvent newMessage, Context context, 
                                 Collector<Tuple2<String, String>> collector) throws Exception {
            
            // 获取当前消息队列
            List<MessageEvent> messageQueue = new ArrayList<>();
            for (MessageEvent message : messageQueueState.get()) {
                messageQueue.add(message);
            }
            
            // 根据优先级插入消息
            if ("HIGH".equals(newMessage.priority)) {
                messageQueue.add(0, newMessage); // 高优先级消息放到队首
            } else {
                messageQueue.add(newMessage); // 普通消息放到队尾
            }
            
            // 限制队列大小
            if (messageQueue.size() > MAX_QUEUE_SIZE) {
                messageQueue.remove(messageQueue.size() - 1); // 移除队尾消息
            }
            
            // 更新状态
            messageQueueState.clear();
            messageQueueState.addAll(messageQueue);
            
            // 处理消息（模拟消费）
            if (!messageQueue.isEmpty()) {
                MessageEvent processedMessage = messageQueue.remove(0);
                collector.collect(new Tuple2<>(processedMessage.userId, 
                    "处理消息: " + processedMessage.content));
                
                // 更新状态
                messageQueueState.clear();
                messageQueueState.addAll(messageQueue);
                
                System.out.println(String.format("📨 处理消息: %s，队列剩余: %d", 
                    processedMessage.content, messageQueue.size()));
            }
        }
    }
    
    /**
     * 模拟用户行为事件源
     */
    public static class MockUserBehaviorEventSource implements SourceFunction<UserBehaviorEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] actions = {"view", "click", "purchase", "add_to_cart"};
        private final String[] pages = {"home", "product", "cart", "checkout"};
        private final String[] products = {"product1", "product2", "product3", "product4", "product5"};
        
        @Override
        public void run(SourceContext<UserBehaviorEvent> ctx) throws Exception {
            while (running) {
                UserBehaviorEvent event = new UserBehaviorEvent(
                    userIds[random.nextInt(userIds.length)],
                    actions[random.nextInt(actions.length)],
                    pages[random.nextInt(pages.length)],
                    products[random.nextInt(products.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(event);
                Thread.sleep(1000 + random.nextInt(2000)); // 1-3秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟购物车商品事件源
     */
    public static class MockCartItemEventSource implements SourceFunction<CartItem> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] productIds = {"prod1", "prod2", "prod3", "prod4", "prod5"};
        private final String[] productNames = {"手机", "电脑", "书籍", "衣服", "食品"};
        
        @Override
        public void run(SourceContext<CartItem> ctx) throws Exception {
            while (running) {
                int index = random.nextInt(productIds.length);
                CartItem item = new CartItem(
                    productIds[index],
                    productNames[index],
                    random.nextInt(3) + 1, // 1-3个商品
                    100.0 + random.nextDouble() * 900.0, // 100-1000随机价格
                    System.currentTimeMillis()
                );
                
                ctx.collect(item);
                Thread.sleep(2000 + random.nextInt(3000)); // 2-5秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 模拟消息事件源
     */
    public static class MockMessageEventSource implements SourceFunction<MessageEvent> {
        private volatile boolean running = true;
        private final Random random = new Random();
        private final String[] userIds = {"user1", "user2", "user3", "user4", "user5"};
        private final String[] priorities = {"HIGH", "NORMAL", "LOW"};
        private final String[] contents = {"订单确认", "支付成功", "发货通知", "系统公告", "活动推送"};
        private int messageIdCounter = 1;
        
        @Override
        public void run(SourceContext<MessageEvent> ctx) throws Exception {
            while (running) {
                MessageEvent message = new MessageEvent(
                    userIds[random.nextInt(userIds.length)],
                    "msg_" + (messageIdCounter++),
                    contents[random.nextInt(contents.length)],
                    priorities[random.nextInt(priorities.length)],
                    System.currentTimeMillis()
                );
                
                ctx.collect(message);
                Thread.sleep(1500 + random.nextInt(2500)); // 1.5-4秒随机间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 运行ListState示例
     */
    public static void runListStateExample(StreamExecutionEnvironment env) throws Exception {
        System.out.println("🚀 开始运行ListState示例...");
        
        // 1. 用户行为轨迹追踪示例
        DataStream<UserBehaviorEvent> behaviorEventStream = env.addSource(new MockUserBehaviorEventSource())
            .name("用户行为事件源");
        
        DataStream<Tuple2<String, List<UserBehaviorEvent>>> behaviorTrackStream = behaviorEventStream
            .keyBy(event -> event.userId)
            .flatMap(new UserBehaviorTracker())
            .name("用户行为轨迹追踪");
        
        behaviorTrackStream.print("行为轨迹");
        
        // 2. 滑动窗口数据缓存示例
        DataStream<Tuple3<String, String, Integer>> slidingWindowStream = behaviorEventStream
            .keyBy(event -> event.userId)
            .process(new SlidingWindowCache())
            .name("滑动窗口数据缓存");
        
        slidingWindowStream.print("滑动窗口统计");
        
        // 3. 购物车商品管理示例
        DataStream<CartItem> cartItemStream = env.addSource(new MockCartItemEventSource())
            .name("购物车商品事件源");
        
        DataStream<Tuple2<String, List<CartItem>>> cartManagementStream = cartItemStream
            .keyBy(item -> item.productId)
            .process(new ShoppingCartManager())
            .name("购物车商品管理");
        
        cartManagementStream.print("购物车管理");
        
        // 4. 消息队列状态管理示例
        DataStream<MessageEvent> messageEventStream = env.addSource(new MockMessageEventSource())
            .name("消息事件源");
        
        DataStream<Tuple2<String, String>> messageQueueStream = messageEventStream
            .keyBy(message -> message.userId)
            .process(new MessageQueueManager())
            .name("消息队列管理");
        
        messageQueueStream.print("消息队列");
        
        System.out.println("✅ ListState示例配置完成");
    }
    
    /**
     * 单独运行ListState示例的主方法
     */
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 配置状态后端
        com.flink.hbase.statemanagement.config.StateBackendConfig
            .configureDevRocksDB(env);
        
        // 运行示例
        runListStateExample(env);
        
        // 执行作业
        env.execute("ListState托管状态示例");
    }
} 